package com.bigdata.kafkastream;

import org.apache.kafka.streams.processor.Processor;
import org.apache.kafka.streams.processor.ProcessorContext;

public class LogProcessor implements Processor<byte[],byte[]> {
    private ProcessorContext context;
    //初始化，连接上下文做传输使用
    @Override
    public void init(ProcessorContext context) {
        this.context = context;
    }
    //
    @Override
    public void process(byte[] key, byte[] value) {
     //MOVIE_RATING_PREFIX:1|1343|2.0|1260759131
        String line = new String(value);

        if(line.contains("MOVIE_RATING_PREFIX:")){
            System.out.println(line);
           line = line.split("MOVIE_RATING_PREFIX:")[1].trim();
            System.out.println(line);
           context.forward("LogProcessor".getBytes(),line.getBytes());
        }
    }

    @Override
    public void punctuate(long l) {

    }

    @Override
    public void close() {

    }
}
